
package com.cx.base.day25_Rocketmq.producer.syncInfo;

import com.cx.base.day25_Rocketmq.order.OrderStep;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.List;

public class SyncDelayProducer {
    public static void main(String[] args) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        //        1.创建消息生产者，并且制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
//        设置超时时间
        producer.setSendMsgTimeout(10000);
//        2.指定namesrv地址
        producer.setNamesrvAddr("192.168.48.128:9876;192.168.48.137:9876");
//        producer.setNamesrvAddr("47.96.165.93:9876;106.52.131.197:9876");
//        3.启动producer
        try {
            producer.start();
        } catch (
                MQClientException e) {
            e.printStackTrace();
        }

        List<OrderStep> orderStepList = OrderStep.buildOrders();

        for (OrderStep order : orderStepList) {
//        4.创建消息对象，指定主题topic,tag和消息体
            Message msg = new Message("delayTopic", "delay", "delayOrder" + order.getOrderId(), order.getDesc().getBytes());
//            RocketMQ 目前只支持固定精度时间的延时消息发送（配置 Message.setDelayTimeLevel 延时精度），默认有18个时间精度的 level，分别是：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
//            这个配置项配置了从1级开始，各级延时的时间，可以修改这个指定级别的延时时间
//            时间单位支持：s、m、h、d，分别表示秒、分、时、天
//            默认值就是上面声明的，可手工调整
//            默认值已够用，不建议修改这个值
            msg.setDelayTimeLevel(3);
//        5.发送消息,参数1：消息对象；参数2：消息队列的选择器；参数3：选择队列的业务标识（orderId）
            SendResult result = producer.send(msg, new MessageQueueSelector() {
                /**
                 *
                 * @param list  消息队列集合
                 * @param message   消息对象
                 * @param o  业务标识
                 * @return
                 */
                @Override
                public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                    long orderId = (long) o;
                    int index = (int) (orderId % list.size());
                    return list.get(index);
                }
            }, order.getOrderId());

            SendStatus status = result.getSendStatus();

            String msgId = result.getMsgId();

            int queueId = result.getMessageQueue().getQueueId();

            System.out.println("send status:" + status + ",message ID:" + msgId + ",queue:" + queueId + ",desc:" + order.getDesc());
        }
//                * 关闭producer
        producer.shutdown();
    }
}
